package com.edu.learning.concurrent.sample.mode4;

import java.util.Random;
import java.util.concurrent.*;

import static java.util.concurrent.TimeUnit.SECONDS;

/**
 * TransferQueue继承自BlockingQueue
 * TransferQueue提供了一个场所，生产者线程使用transfer方法传入一些对象并阻塞，直至这些对象被消费者线程全部取出。
 * SynchronousQueue很像一个容量为0的TransferQueue。
 * TransferQueue共包括以下方法：
 * 1. transfer(E e)，若当前存在一个正在等待获取的消费者线程，即立刻移交之；否则会将元素e插入到队列尾部，并进入阻塞状态，直到有消费者线程取走该元素。
 * 2. tryTransfer(E e)，若当前存在一个正在等待获取的消费者线程（使用take()或者poll()函数），即立刻移交之； 否则返回false，并且不进入队列，这是一个非阻塞的操作。
 * 3. tryTransfer(E e, long timeout, TimeUnit unit) 若当前存在一个正在等待获取的消费者线程，即立刻移交之；否则会将元素e插入到队列尾部，并且等待被消费者线程获取消费掉，若在指定的时间内元素e无法被消费者线程获取，则返回false，同时该元素被移除。
 * 4. hasWaitingConsumer() 判断是否存在消费者线程。
 * 5. getWaitingConsumerCount() 获取所有等待获取元素的消费线程数量。
 *
 * TransferQueue是一个聪明的队列，它是ConcurrentLinkedQueue, SynchronousQueue (在公平模式下), 无界的LinkedBlockingQueues等的超集。
 * 所以，在合适的场景中，请尽量使用TransferQueue
 *
 * 例子，一个生产者使用transfer方法传输10个字符串，两个消费者线程则各取出5个字符串，可以看到生产者在transfer时会一直阻塞直到所有字符串被取出：
 * @author ll
 * @Date 2017/10/26 15:07
 */
public class TransferQueueDemo {
    public static void main(String[] args) {
        TransferQueue<String> queue = new LinkedTransferQueue<>();
        ExecutorService executorService = new ThreadPoolExecutor(
                16,
                16,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1024),
                new ThreadPoolExecutor.CallerRunsPolicy());
        executorService.submit(new Producer("Producer1 ", queue));
        executorService.submit(new Consumer("Consumer1 ", queue));
        executorService.submit(new Consumer("Consumer2 ", queue));
        //executorService.shutdown();
    }

    static class Producer implements Runnable {
        private final String name;
        private final TransferQueue<String> queue;

        public Producer(String name, TransferQueue<String> queue) {
            this.name = name;
            this.queue = queue;
        }

        @Override
        public void run() {
            System.out.println("begin transfer objects");
            try {
                for (int i = 0; i < 10; i ++) {
                    queue.transfer("Product " + i);
                    System.out.println(name + " transfer " + " Product " + i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class Consumer implements Runnable {

        private final String name;
        private final TransferQueue<String> queue;
        private static Random random = new Random(System.currentTimeMillis());

        public Consumer(String name, TransferQueue<String> queue) {
            this.name = name;
            this.queue = queue;
        }

        @Override
        public void run() {
            try {
                for (int i = 0; i < 5; i ++) {
                    String str = queue.take();
                    System.out.println(name + " take " + str);
                    SECONDS.sleep(random.nextInt(5));
                }
                System.out.println(name + " is over");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
